package com.atguigu.day03;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.Random;

public class Flink01_Source_Customer {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        //TODO 2.通过自定义Source获取数据
        DataStreamSource<WaterSensor> streamSource = env.addSource(new MySource()).setParallelism(2);

        streamSource.print();

        env.execute();

    }

    //自定义一个类实现SourceFun接口
//    public static class MySource implements SourceFunction<WaterSensor>{
    //如果要实现设置多并行度可以实现ParallelSourceFunction这个接口
    public static class MySource implements ParallelSourceFunction<WaterSensor> {
        private Random random = new Random();

        //定义标志位，用来控制while循环
        private Boolean isRunning = true;

        /**
         *  主要用来生成数据的，通常这个方法中都有一个while循环，然后将生产的数据通过SourceContext发送出去
         * @param ctx
         * @throws Exception
         */
        @Override
        public void run(SourceContext<WaterSensor> ctx) throws Exception {
            while (isRunning){
                ctx.collect(new WaterSensor("sensor_"+random.nextInt(100), System.currentTimeMillis(), random.nextInt(1000)));

                Thread.sleep(1000);
            }
        }

        /**
         * 用来将run方法中的while循环跳出，进而终止生成数据
         */
        @Override
        public void cancel() {
            isRunning = false;
        }
    }
}
